package cn.itcast.b_etl.transformation;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 案例:实时单词统计
 */
public class KeyByDemo {

    public static void main(String[] args) throws Exception {

        /**
         * 开发步骤
         * 1.获取流处理运行环境
         * 2.获取数据源socket
         * 3.数据分组
         * 4.数据求和
         * 5.数据打印
         * 6.触发执行
         */
        //1.获取流处理运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.接收实时数据
        DataStreamSource<String> source = env.socketTextStream("node1", 8090);
        //需求单词统计
        source.map(new MapFunction<String, Tuple2<String ,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        return Tuple2.of(value,1);
                    }
                })
                .keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0;
                    }
                }) //在流处理中，分组用keyBy
//                .keyBy(line->line.f0)  //lambda表达式用法，代码简洁，不方便维护和调试
//                .sum(1)
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                        return Tuple2.of(value1.f0,value1.f1+value2.f1);
                    }
                })
                .print();  //4.数据输出 （print,不是触发算子）


        //执行程序
        env.execute();
    }


}
